package beautiful.butterfly.drds.data_exchange.transport.transformer;


import beautiful.butterfly.drds.data_exchange.exception.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.transport.record.column.*;
import org.apache.commons.lang3.StringUtils;


public class FilterTransformer extends Transformer
{
    public FilterTransformer()
    {
        setTransformerName("dx_filter");
    }


    public Record evaluate(Record record, Object... paras)
    {

        int columnIndex;
        String expression;
        String value;

        if (paras.length != 3)
        {
            throw new IllegalArgumentException("dx_filter paras must be 3");
        }
        columnIndex = (Integer) paras[0];
        expression = (String) paras[1];
        value = (String) paras[2];

        if (StringUtils.isEmpty(value))
        {
            throw new IllegalArgumentException("dx_filter para 2 can't be null");
        }


        Column column = record.getColumn(columnIndex);

        try
        {

            if (expression.equalsIgnoreCase("like"))
            {
                return doLike(record, value, column);
            } else if (expression.equalsIgnoreCase("not like"))
            {
                return doNotLike(record, value, column);
            } else if (expression.equalsIgnoreCase(">"))
            {
                return doGreat(record, value, column, false);
            } else if (expression.equalsIgnoreCase("<"))
            {
                return doLess(record, value, column, false);
            } else if (expression.equalsIgnoreCase("=") || expression.equalsIgnoreCase("=="))
            {
                return doEqual(record, value, column);
            } else if (expression.equalsIgnoreCase("!="))
            {
                return doNotEqual(record, value, column);
            } else if (expression.equalsIgnoreCase(">="))
            {
                return doGreat(record, value, column, true);
            } else if (expression.equalsIgnoreCase("<="))
            {
                return doLess(record, value, column, true);
            } else
            {
                throw new RuntimeException("dx_filter can't suport code:" + expression);
            }
        } catch (Exception e)
        {
            throw DataExchangeException.asDataExchangeException(TransformerErrorCode.TRANSFORMER_RUN_EXCEPTION, e.getMessage(), e);
        }
    }


    private Record doGreat(Record record, String value, Column column, boolean hasEqual)
    {

        //如果字段为空，直接不参与比较。即空也属于无穷小
        if (column.getObject() == null)
        {
            return record;
        }
        if (column instanceof DoubleColumn)
        {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (hasEqual)
            {
                if (ori >= val)
                {
                    return null;
                } else
                {
                    return record;
                }
            } else
            {
                if (ori > val)
                {
                    return null;
                } else
                {
                    return record;
                }
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn
                )
        {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (hasEqual)
            {
                if (ori >= val)
                {
                    return null;
                } else
                {
                    return record;
                }
            } else
            {
                if (ori > val)
                {
                    return null;
                } else
                {
                    return record;
                }
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BooleanColumn)
        {
            String ori = column.asString();
            if (hasEqual)
            {
                if (ori.compareTo(value) >= 0)
                {
                    return null;
                } else
                {
                    return record;
                }
            } else
            {
                if (ori.compareTo(value) > 0)
                {
                    return null;
                } else
                {
                    return record;
                }
            }
        } else
        {
            throw new RuntimeException(">=,> can't support this columnType:" + column.getClass().getSimpleName());
        }
    }

    private Record doLess(Record record, String value, Column column, boolean hasEqual)
    {

        //如果字段为空，直接不参与比较。即空也属于无穷大
        if (column.getObject() == null)
        {
            return record;
        }

        if (column instanceof DoubleColumn)
        {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (hasEqual)
            {
                if (ori <= val)
                {
                    return null;
                } else
                {
                    return record;
                }
            } else
            {
                if (ori < val)
                {
                    return null;
                } else
                {
                    return record;
                }
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn)
        {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (hasEqual)
            {
                if (ori <= val)
                {
                    return null;
                } else
                {
                    return record;
                }
            } else
            {
                if (ori < val)
                {
                    return null;
                } else
                {
                    return record;
                }
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BooleanColumn)
        {
            String ori = column.asString();
            if (hasEqual)
            {
                if (ori.compareTo(value) <= 0)
                {
                    return null;
                } else
                {
                    return record;
                }
            } else
            {
                if (ori.compareTo(value) < 0)
                {
                    return null;
                } else
                {
                    return record;
                }
            }
        } else
        {
            throw new RuntimeException("<=,< can't support this columnType:" + column.getClass().getSimpleName());
        }

    }

    /**
     * DateColumn将比较long值，StringColumn，ByteColumn以及BooleanColumn比较其String值
     *
     * @param record
     * @param value
     * @param column
     * @return 如果相等，则过滤。
     */

    private Record doEqual(Record record, String value, Column column)
    {

        //如果字段为空，只比较目标字段为"null"，否则null字段均不过滤
        if (column.getObject() == null)
        {
            if (value.equalsIgnoreCase("null"))
            {
                return null;
            } else
            {
                return record;
            }
        }

        if (column instanceof DoubleColumn)
        {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (ori == val)
            {
                return null;
            } else
            {
                return record;
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn)
        {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (ori == val)
            {
                return null;
            } else
            {
                return record;
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BooleanColumn)
        {
            String ori = column.asString();
            if (ori.compareTo(value) == 0)
            {
                return null;
            } else
            {
                return record;
            }
        } else
        {
            throw new RuntimeException("== can't support this columnType:" + column.getClass().getSimpleName());
        }

    }

    /**
     * DateColumn将比较long值，StringColumn，ByteColumn以及BooleanColumn比较其String值
     *
     * @param record
     * @param value
     * @param column
     * @return 如果不相等，则过滤。
     */
    private Record doNotEqual(Record record, String value, Column column)
    {

        //如果字段为空，只比较目标字段为"null", 否则null字段均过滤。
        if (column.getObject() == null)
        {
            if (value.equalsIgnoreCase("null"))
            {
                return record;
            } else
            {
                return null;
            }
        }

        if (column instanceof DoubleColumn)
        {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (ori != val)
            {
                return null;
            } else
            {
                return record;
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn)
        {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (ori != val)
            {
                return null;
            } else
            {
                return record;
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BooleanColumn)
        {
            String ori = column.asString();
            if (ori.compareTo(value) != 0)
            {
                return null;
            } else
            {
                return record;
            }
        } else
        {
            throw new RuntimeException("== can't support this columnType:" + column.getClass().getSimpleName());
        }
    }

    private Record doLike(Record record, String value, Column column)
    {
        String orivalue = column.asString();
        if (orivalue != null && orivalue.matches(value))
        {
            return null;
        } else
        {
            return record;
        }
    }

    private Record doNotLike(Record record, String value, Column column)
    {
        String orivalue = column.asString();
        if (orivalue != null && orivalue.matches(value))
        {
            return record;
        } else
        {
            return null;
        }
    }
}
